package com.topsun.framework.common.kafka.producer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.topsun.framework.common.core.util.IdUtil;
import com.topsun.framework.common.kafka.pojo.Message;
import com.topsun.framework.common.kafka.pojo.MessageResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Package cloud.chenglang.data.kafka
 * @author: 皮皮
 * @date: 2021/7/8 10:18 AM
 */
@Slf4j
public abstract class DefaultProducer<K, V> implements InitializingBean {


    private KafkaTemplate<K, String> kafkaTemplate;

    private String topic;

    public void setKafkaTemplate(KafkaTemplate<K, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public KafkaTemplate<K, String> getKafkaTemplate() {
        return kafkaTemplate;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    public void afterPropertiesSet() {
        if (StrUtil.isEmpty(getTopic())) {
            throw new KafkaException("can not without topic");
        }
    }

    private ListenableFuture<SendResult<K, String>> send(Message<K, V> message) {
        if (StrUtil.isEmpty(getTopic())) {
            throw new KafkaException("can not without topic");
        }
        message.setMessageId(IdUtil.getInstance().getId());
        String messageStr = JSONObject.toJSONString(message);
        List<Header> recordHeaderList = message.getHeaders().entrySet().stream().map(entry -> new RecordHeader(entry.getKey(), StrUtil.bytes(entry.getValue()))).collect(Collectors.toList());
        ProducerRecord<K, String> producerRecord = new ProducerRecord<K, String>(getTopic(), null, null, message.getKey(), messageStr, recordHeaderList);
        return getKafkaTemplate().send(producerRecord);
    }

    /**
     * 同步发送消息
     * @param data
     */
    public void syncSend(V data) {
        syncSend(null, data);
    }

    public void syncSend(K key, V data) {
        syncSend(Message.<K, V>builder().key(key).data(data).build());
    }

    /**
     * 同步发送消息
     * @param message
     */
    public void syncSend(Message<K, V> message) {
        ListenableFuture<SendResult<K, String>> future = send(message);
        try {
            SendResult<K, String> sendResult = future.get();
            log.debug("sync send message success, topic is{}, value is {}", sendResult.getProducerRecord().topic(), sendResult.getProducerRecord().value());
            doSendAfter(sendResult);
        } catch (Exception  e) {
            throw new KafkaException("sync send message error", e);
        }
    }

    /**
     * 异步发送
     * @param message
     */
    public void asyncSend(V message) {
        asyncSend(null, message);
    }

    /**
     * 异步发送
     * @param key
     * @param data
     */
    public void asyncSend(K key, V data) {
        asyncSend(Message.<K, V>builder().key(key).data(data).build());
    }

    public void asyncSend(Message<K, V> message) {
        asyncSend(message, new DefaultCallback());
    }

    /**
     * 异步发送
     * @param message
     * @param listenableFutureCallback
     */
    public void asyncSend(Message<K, V> message, ListenableFutureCallback<SendResult<K, String>> listenableFutureCallback) {
        ListenableFuture<SendResult<K, String>> future = send(message);
        future.addCallback(listenableFutureCallback);
    }

    private void doSendAfter(SendResult<K, String> sendResult) {
        ProducerRecord<K, String> producerRecord = sendResult.getProducerRecord();
        RecordMetadata recordMetadata = sendResult.getRecordMetadata();
        MessageResult<K, V> messageResult = MessageResult.<K, V>builder().topic(getTopic())
                .message(JSON.parseObject(producerRecord.value(), ParameterizedTypeImpl.make(Message.class, getMessageGenericType(), Message.class)))
                .partition(recordMetadata.partition()).offset(recordMetadata.offset()).timestamp(recordMetadata.timestamp()).build();
        doSendAfter(messageResult);
    }


    /**
     * 发送结果处理
     * @param messageResult
     */
    public void doSendAfter(MessageResult<K, V> messageResult) {}

    /**
     * 获取泛型数据类型
     * @return
     */
    private Type[] getMessageGenericType() {
        Type superType = this.getClass().getGenericSuperclass();
        return (superType instanceof ParameterizedType ? ((ParameterizedType)superType).getActualTypeArguments() : new Type[]{});
    }

    class DefaultCallback implements ListenableFutureCallback<SendResult<K, String>> {
        @Override
        public void onFailure(Throwable ex) {
            log.error("async send mq msg error", ex);
        }

        @Override
        public void onSuccess(SendResult<K, String> sendResult) {
            log.debug("async send mq msg success: {}", JSON.toJSONString(sendResult));
            doSendAfter(sendResult);
        }
    }

}
